// TripDataMapper.java - 修正版
package com.taxi.analysis.fraud;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

/**
 * 修正版行程数据Mapper
 * 功能：处理行程数据，为Join操作准备数据
 * 技术：Join + 条件过滤
 */
public class TripDataMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();
    private String inputFileName;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 获取输入文件名，用于区分不同的数据源
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
        inputFileName = fileSplit.getPath().getName();
        System.out.println("处理文件: " + inputFileName);
    }

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        // 跳过CSV标题行
        if (key.get() == 0) {
            return;
        }

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) {
                return;
            }

            // 根据文件名判断数据类型
            if (inputFileName.contains("trips")) {
                processTripData(line, context);
            } else if (inputFileName.contains("locations")) {
                processLocationData(line, context);
            }

        } catch (Exception e) {
            System.err.println("处理行时出错: " + value.toString());
            System.err.println("错误信息: " + e.getMessage());
        }
    }

    /**
     * 处理行程数据
     */
    private void processTripData(String line, Context context)
            throws IOException, InterruptedException {

        String[] fields = line.split(",");

        // 行程数据字段索引:
        // 0:trip_id, 1:PULocationID, 2:DOLocationID, 3:trip_distance, 4:total_amount,
        // 5:passenger_count, 6:trip_duration, 7:avg_speed, 8:fare_per_mile,
        // 9:pickup_hour, 10:speed_anomaly, 11:fare_anomaly, 12:ratio_anomaly,
        // 13:duration_anomaly, 14:is_suspicious

        if (fields.length >= 15) {
            try {
                String tripId = fields[0].trim();
                String pickupLocationId = fields[1].trim();
                String dropoffLocationId = fields[2].trim();
                double tripDistance = parseDouble(fields[3]);
                double totalAmount = parseDouble(fields[4]);
                int passengerCount = parseInt(fields[5]);
                double tripDuration = parseDouble(fields[6]);
                double avgSpeed = parseDouble(fields[7]);
                double farePerMile = parseDouble(fields[8]);
                int pickupHour = parseInt(fields[9]);
                boolean speedAnomaly = parseBoolean(fields[10]);
                boolean fareAnomaly = parseBoolean(fields[11]);
                boolean ratioAnomaly = parseBoolean(fields[12]);
                boolean durationAnomaly = parseBoolean(fields[13]);
                boolean isSuspicious = parseBoolean(fields[14]);

                // 条件过滤：只处理有效的行程数据
                if (isValidTrip(tripId, pickupLocationId, dropoffLocationId, tripDistance, totalAmount)) {

                    // 为上车地点生成Join键值对
                    String pickupJoinKey = "PICKUP_" + pickupLocationId;
                    String tripInfo = String.format("TRIP|%s|PICKUP|%s|%s|%.2f|%.2f|%d|%.2f|%.2f|%.2f|%d|%s|%s|%s|%s|%s",
                            tripId, pickupLocationId, dropoffLocationId, tripDistance, totalAmount,
                            passengerCount, tripDuration, avgSpeed, farePerMile, pickupHour,
                            speedAnomaly ? "1" : "0", fareAnomaly ? "1" : "0",
                            ratioAnomaly ? "1" : "0", durationAnomaly ? "1" : "0",
                            isSuspicious ? "1" : "0");

                    outputKey.set(pickupJoinKey);
                    outputValue.set(tripInfo);
                    context.write(outputKey, outputValue);

                    // 为下车地点生成Join键值对
                    String dropoffJoinKey = "DROPOFF_" + dropoffLocationId;
                    tripInfo = String.format("TRIP|%s|DROPOFF|%s|%s|%.2f|%.2f|%d|%.2f|%.2f|%.2f|%d|%s|%s|%s|%s|%s",
                            tripId, pickupLocationId, dropoffLocationId, tripDistance, totalAmount,
                            passengerCount, tripDuration, avgSpeed, farePerMile, pickupHour,
                            speedAnomaly ? "1" : "0", fareAnomaly ? "1" : "0",
                            ratioAnomaly ? "1" : "0", durationAnomaly ? "1" : "0",
                            isSuspicious ? "1" : "0");

                    outputKey.set(dropoffJoinKey);
                    outputValue.set(tripInfo);
                    context.write(outputKey, outputValue);
                }
            } catch (Exception e) {
                System.err.println("解析行程数据错误: " + line);
            }
        }
    }

    /**
     * 处理地点数据
     */
    private void processLocationData(String line, Context context)
            throws IOException, InterruptedException {

        String[] fields = line.split(",");

        // 地点数据字段索引:
        // 0:LocationID, 1:total_trips, 2:suspicious_trips, 3:suspicious_rate,
        // 4:avg_fare, 5:std_fare, 6:avg_distance, 7:std_distance,
        // 8:avg_speed_stat, 9:std_speed, 10:avg_fare_per_mile, 11:std_fare_per_mile,
        // 12:location_type, 13:risk_level

        if (fields.length >= 14) {
            try {
                String locationId = fields[0].trim();
                int totalTrips = parseInt(fields[1]);
                int suspiciousTrips = parseInt(fields[2]);
                double suspiciousRate = parseDouble(fields[3]);
                double avgFare = parseDouble(fields[4]);
                double stdFare = parseDouble(fields[5]);
                double avgDistance = parseDouble(fields[6]);
                double stdDistance = parseDouble(fields[7]);
                double avgSpeedStat = parseDouble(fields[8]);
                double stdSpeed = parseDouble(fields[9]);
                double avgFarePerMile = parseDouble(fields[10]);
                double stdFarePerMile = parseDouble(fields[11]);
                String locationType = fields[12].trim();
                String riskLevel = fields[13].trim();

                // 条件过滤：只处理有足够数据的地点
                if (isValidLocation(locationId, totalTrips, locationType)) {

                    // 生成Join键
                    String joinKey = locationType.toUpperCase() + "_" + locationId;

                    // 创建地点统计信息
                    String locationInfo = String.format("LOCATION|%s|%d|%d|%.3f|%.2f|%.2f|%.2f|%.2f|%.2f|%.2f|%.2f|%.2f|%s|%s",
                            locationId, totalTrips, suspiciousTrips, suspiciousRate, avgFare,
                            stdFare, avgDistance, stdDistance, avgSpeedStat, stdSpeed,
                            avgFarePerMile, stdFarePerMile, locationType, riskLevel);

                    outputKey.set(joinKey);
                    outputValue.set(locationInfo);
                    context.write(outputKey, outputValue);
                }
            } catch (Exception e) {
                System.err.println("解析地点数据错误: " + line);
            }
        }
    }

    /**
     * 验证行程数据的有效性
     */
    private boolean isValidTrip(String tripId, String pickupLocationId, String dropoffLocationId,
                                double tripDistance, double totalAmount) {
        return !tripId.isEmpty() &&
                !pickupLocationId.isEmpty() &&
                !dropoffLocationId.isEmpty() &&
                !tripId.equals("null") &&
                !pickupLocationId.equals("null") &&
                !dropoffLocationId.equals("null") &&
                tripDistance >= 0 &&
                totalAmount > 0;
    }

    /**
     * 验证地点数据的有效性
     */
    private boolean isValidLocation(String locationId, int totalTrips, String locationType) {
        return !locationId.isEmpty() &&
                !locationType.isEmpty() &&
                !locationId.equals("null") &&
                !locationType.equals("null") &&
                totalTrips >= 1; // 至少有1次行程的地点
    }

    /**
     * 安全的double解析
     */
    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    /**
     * 安全的int解析
     */
    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }

    /**
     * 安全的boolean解析
     */
    private boolean parseBoolean(String str) {
        try {
            return "1".equals(str.trim()) ||
                    "true".equalsIgnoreCase(str.trim()) ||
                    "True".equals(str.trim());
        } catch (Exception e) {
            return false;
        }
    }
}
